package com.codejiwei.flink.practice;

import com.codejiwei.flink.entity.WaterSensor;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * @author jiwei
 * @description
 * @date 2023/5/22 11:23
 */
public class Flink_Transform_Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        KeyedStream<WaterSensor, String> kbStream = env
                .fromCollection(waterSensors)
                .keyBy(WaterSensor::getId);

//        kbStream.reduce(new ReduceFunction<WaterSensor>() {
//            @Override
//            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
//                System.out.println("reduce function....");
//                return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
//            }
//        }).print("reduce...");

        kbStream.reduce((value1, value2) -> {
            return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
        }).print();

        env.execute();

    }
}
